package com.lsx143.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.lsx143.realtime.app.BaseAppV2;
import com.lsx143.realtime.bean.OrderDetail;
import com.lsx143.realtime.bean.OrderInfo;
import com.lsx143.realtime.bean.OrderWide;
import com.lsx143.realtime.common.Constants;
import com.lsx143.realtime.function.DimAsyncFunction;
import com.lsx143.realtime.util.DimUtil;
import com.lsx143.realtime.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * DWM-订单维度宽表
 */
public class DWMOrderWideAppAsyncCache extends BaseAppV2 {
    public static void main(String[] args) {
        new DWMOrderWideAppAsyncCache().init(
                20002,
                "DWMOrderWideAppAsyncCache",
                1,
                "DWMOrderWideAppAsyncCache",
                Constants.TOPIC_DWD_ORDER_INFO,
                Constants.TOPIC_DWD_ORDER_DETAIL);
    }

    /**
     * 生成订单维度宽表的逻辑
     *
     * @param env        flink的执行环境
     * @param srcStreams 多个源数据流
     */
    @Override
    protected void run(StreamExecutionEnvironment env, Map<String, DataStreamSource<String>> srcStreams) {
        System.out.println("【DWMOrderWideAppAsyncCache】订单宽表App启动");
        //1.读取order_info和order_detail事实表
        SingleOutputStreamOperator<OrderWide> orderWideWithoutDimsStream = getOrderWideWithoutDims(srcStreams);

        //2.添加其他维度属性到宽表中
        SingleOutputStreamOperator<OrderWide> orderWideWithDimsStream = getOrderWideWithDims(orderWideWithoutDimsStream);

        //3.写出到kafka
        writeToKafka(orderWideWithDimsStream);

    }

    /**
     * 写出到kafka
     *
     * @param orderWideWithDimsStream 宽表流
     */
    private void writeToKafka(SingleOutputStreamOperator<OrderWide> orderWideWithDimsStream) {
        System.out.println("准备写出到kafka");
        orderWideWithDimsStream
                .map(JSONObject::toJSONString)
                .addSink(KafkaUtil.getKafkaSink(Constants.TOPIC_DWM_ORDER_WIDE));
    }

    /**
     * 将DIM维度表信息添加到宽表
     * 这种连接方式下有个弊端就是,没有一个订单都会去数据库查询,效率比较低
     *
     * @param orderWideWithoutDimsStream 订单与订单详情事实表join后的数据流
     * @return 订单数据宽表
     */
    private SingleOutputStreamOperator<OrderWide> getOrderWideWithDims(SingleOutputStreamOperator<OrderWide> orderWideWithoutDimsStream) {
        return AsyncDataStream.unorderedWait(
                orderWideWithoutDimsStream,
                new DimAsyncFunction<OrderWide>() {
                    @Override
                    public void timeout(OrderWide input, ResultFuture<OrderWide> resultFuture) {
                        throw new RuntimeException(" 异步超时:\n" +
                                " 1. 检测你的集群是否全部开启: redis Phoenix hadoop ...\n" +
                                " 2. 检测在Phoenix中, 六张维度表数据是否都在\n" +
                                " 3. 检测dim_user_info的数据是否4000\n" +
                                " 4. 检测一些redis是否允许远程连接\n" +
                                "            redis-server /etc/redis.conf\n" +
                                " 5. 检测redis中维度的缓存数据中列名的大写问题\n" +
                                " 6. 找我");
                    }

                    @Override
                    protected void addDim(Connection phoenixCon,
                                          Jedis jedis,
                                          OrderWide orderWide,
                                          ResultFuture<OrderWide> resultFuture) {
                        //1.用户信息
                        JSONObject userInfo = DimUtil.readDim(phoenixCon, jedis, Constants.DIM_USER_INFO, orderWide.getUser_id());
                        orderWide.setUser_gender(userInfo.getString("GENDER"));
                        orderWide.calcUserAgeByBirthday(userInfo.getString("BIRTHDAY"));

                        //2.省份
                        JSONObject baseProvince = DimUtil.readDim(phoenixCon, jedis, Constants.DIM_BASE_PROVINCE, orderWide.getProvince_id());
                        orderWide.setProvince_name(baseProvince.getString("NAME"));
                        orderWide.setProvince_area_code(baseProvince.getString("AREA_CODE"));
                        orderWide.setProvince_iso_code(baseProvince.getString("ISO_CODE"));
                        orderWide.setProvince_3166_2_code(baseProvince.getString("ISO_3166_2"));

                        //3.sku
                        JSONObject skuInfo = DimUtil.readDim(phoenixCon, jedis, Constants.DIM_SKU_INFO, orderWide.getSku_id());
                        orderWide.setSpu_id(skuInfo.getLong("SPU_ID"));
                        orderWide.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));
                        orderWide.setTm_id(skuInfo.getLong("TM_ID"));
                        orderWide.setOrder_price(skuInfo.getBigDecimal("PRICE"));
                        orderWide.setSku_name(skuInfo.getString("SKU_NAME"));

                        //4.spu
                        JSONObject spuInfo = DimUtil.readDim(phoenixCon, jedis, Constants.DIM_SPU_INFO, orderWide.getSpu_id());
                        orderWide.setSpu_name(spuInfo.getString("SPU_NAME"));

                        //5.目录
                        JSONObject category3 = DimUtil.readDim(phoenixCon, jedis, Constants.DIM_BASE_CATEGORY3, orderWide.getCategory3_id());
                        orderWide.setCategory3_name(category3.getString("NAME"));

                        //6.品牌
                        JSONObject tm = DimUtil.readDim(phoenixCon, jedis, Constants.DIM_BASE_TRADEMARK, orderWide.getTm_id());
                        orderWide.setTm_name(tm.getString("TM_NAME"));

                        //将宽表输出到流
                        resultFuture.complete(Collections.singletonList(orderWide));
                    }
                },
                30,
                TimeUnit.SECONDS
        );
    }

    /**
     * 获取order_info和order_detail事实表
     *
     * @param srcStreams 输入数据流map
     * @return order_info和order_detail宽表
     */
    private SingleOutputStreamOperator<OrderWide> getOrderWideWithoutDims(Map<String, DataStreamSource<String>> srcStreams) {
        //1.读取两个流
        //  1.1 读取OrderInfo
        KeyedStream<OrderInfo, Long> keyedOrderInfoStream = srcStreams
                .get(Constants.TOPIC_DWD_ORDER_INFO)
                .map(json -> JSON.parseObject(json, OrderInfo.class))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((info, ts) -> info.getCreate_ts())
                )
                .keyBy(OrderInfo::getId);


        //  1.2 读取OrderDetail
        KeyedStream<OrderDetail, Long> keyedOrderDetailStream = srcStreams
                .get(Constants.TOPIC_DWD_ORDER_DETAIL)
                .map(json -> JSON.parseObject(json, OrderDetail.class))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((detail, ts) -> detail.getCreate_ts())
                )
                .keyBy(OrderDetail::getOrder_id);
        //2.进行join
        return keyedOrderInfoStream
                .intervalJoin(keyedOrderDetailStream)
                .between(Time.seconds(-5), Time.seconds(5))
                .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                    @Override
                    public void processElement(OrderInfo info,
                                               OrderDetail detail,
                                               Context ctx,
                                               Collector<OrderWide> out) {
                        out.collect(new OrderWide(info, detail));
                    }
                });
    }
}
